package com.learn.lb.spark.sql.work

import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable.ArrayBuffer

/**
 * 协同过滤，倒排式求相似度
 *
 * @author laibo
 * @since 2019/8/12 22:02
 *
 */
object ItemCfSpark {

  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[2]").setAppName("ItemCfSparkTask")
    val spark = new SparkContext(sparkConf)
    val inputPath = "E:\\idea\\workspace\\bd-learn\\spark-learn\\src\\main\\resources\\train_new.data"
    val outputPath = "E:\\idea\\workspace\\bd-learn\\spark-learn\\src\\main\\resources\\train_new_cf_res2"
    val dataRdd = spark.textFile(inputPath)
    // step1.评分归一化，求UI矩阵
    val uiMatrix = dataRdd.filter(_.split("\t").length > 2)
      .map(m => {
        val fields = m.split("\t")
        val uId = fields(0)
        val itemId = fields(1)
        val score = fields(2).toDouble
        (itemId, (uId, score))
      }).groupByKey()
      .flatMap(fm => {
        val itemId = fm._1
        val userScoreArr = fm._2.toArray
        var sumScore = 0D
        //这里求sumScore有点不明白
        //A用户对item1进行了评分，B用户对item1没评分
        //A用户对item2进行了评分, B用户对item2也进行了评分
        //那计算item1和item2的相似度不应该把A用户的评分给过滤掉吗？
        //虽然后面做内积的时候求的同一个用户的评分，但是这里求归一化分母明显会变大吧
        for (us <- userScoreArr) {
          sumScore += us._2 * us._2
        }
        sumScore = Math.sqrt(sumScore)
        //归一化之后返回 （userId, (itemId, score)）
        //这样做是为了方便分组，注意，score是归一化之后的结果, 也就是求单位向量
        val resultArray = new ArrayBuffer[(String, (String, Double))]()
        for (us <- userScoreArr) {
          resultArray += ((us._1, (itemId, us._2 / sumScore)))
        }
        resultArray
      }).groupByKey()
    //衍生II Pair对，key 为两个itemId，value为评分
    //因为评分是已经做过归一化的,模是1，直接求内积即可
    val iiPair = uiMatrix.flatMap(fm => {
      val userId = fm._1
      val itemScoreArr = fm._2.toArray
      val resultArray = new ArrayBuffer[((String, String), Double)]()
      val length = itemScoreArr.length
      //两两结合
      for (i <- 0 until length - 1) {
        val firstIemScore = itemScoreArr(i)
        for (j <- i + 1 until length) {
          val secondItemScore = itemScoreArr(j)
          //这里是求内积的过程，只是计算一个用户（也就是一个坐标（x1 * y1））对两个item评分
          //一个用户代表一个坐标，最后需要在将每个用户的评分相加
          resultArray += (((firstIemScore._1, secondItemScore._1), firstIemScore._2 * secondItemScore._2))
          resultArray += (((secondItemScore._1, firstIemScore._1), firstIemScore._2 * secondItemScore._2))
        }
      }
      resultArray
    }).groupByKey()
    //生成结果, 求内积，因为前面每一个坐标都相乘了
    //这里直接做加法就行了
    val iiSimilarScore = iiPair.map(m => {
      val item = m._1
      val scoreArr = m._2.toArray
      var sumScore = 0D
      for (score <- scoreArr) {
        sumScore += score
      }
      (item._1, (item._2, sumScore))
    })
    //取相似度前5
    iiSimilarScore.groupByKey().map(m=>{
      val item1 = m._1
      val itemScoreArr = m._2.toArray.sortWith(_._2 > _._2)
      val limit5 = if (itemScoreArr.length > 5)  5 else itemScoreArr.length
      val resBuild = new StringBuilder()
      resBuild.append(item1)
      for(i <- 0 until limit5){
        val itemScore = itemScoreArr(i)
        resBuild.append("\t")
        resBuild.append(itemScore._1).append(":").append(itemScore._2.formatted("%.2f"))
      }
      resBuild.toString()
    }).saveAsTextFile(outputPath)
  }
}
